package com.cw.realtime.common
package util

import base.FlinkRunner.SqlRunnerConfig
import constant.Constant._

import io.debezium.util.Clock.system

object SqlUtil {


  def getKafkaSourceSql(topic: String, groupId: String): String = {

    s"""WITH (
  'connector' = 'kafka',
  'topic' = '$topic',
  'properties.bootstrap.servers' = '$KAFKA_BROKERS',
  'properties.group.id' = '$groupId',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
)
    """
  }


  def getKafkaSourceSql(topic: String)(implicit config: SqlRunnerConfig): String = {

    s"""WITH (
  'connector' = 'kafka',
  'topic' = '$topic',
  'properties.bootstrap.servers' = '$KAFKA_BROKERS',
  'properties.group.id' = '${config.groupId}',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
)
    """
  }


  def getKafkaSinkSql(topic: String): String = {
    s"""WITH (
  'connector' = 'kafka',
  'topic' = '$topic',
  'properties.bootstrap.servers' = '$KAFKA_BROKERS',
  'format' = 'json'
)"""

  }


  def getUpsertKafkaSinkSql(topic: String) = {

    s"""
WITH (
  'connector' = 'upsert-kafka',
  'topic' = '$topic',
  'properties.bootstrap.servers' = '$KAFKA_BROKERS',
  'key.format' = 'json',
  'value.format' = 'json'
)
    """
  }


  def getKafkaTopicDb(groupId: String): String = {

    s"""
CREATE TABLE topic_db (
  `database` string,
  `table` string,
  `type` string,
  `ts` bigint,
  `data` map<string, string>,
  `old` map<string, string>,
  `proc_time` as PROCTIME(),
  raw_time as TO_TIMESTAMP_LTZ(ts * 1000, 3),
  WATERMARK FOR raw_time AS raw_time - INTERVAL '5' SECOND
) ${getKafkaSourceSql(TOPIC_DB, groupId)}
"""

  }

  def getDorisSinkSql(tableName: String): String = {
    s"""
WITH (
 'connector' = 'doris',
 'fenodes' = '$FENODES',
 'table.identifier' = '$DORIS_DB.$tableName',
 'username' = '$DORIS_USERNAME',
 'password' = '$DORIS_PASSWD',
 'sink.label-prefix' = 'doris_label${system.currentTimeInMillis()}'
)
    """




  }


}
